package com.bw.gmall.realtime.app.dws;

import com.bw.gmall.realtime.app.func.SplitFunction;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception{

        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //todo 2.使用ddl方式读取kafka_page_log 主题的数据创建表并且提取时间戳生成Watermark
        String topic = "dwd_traffic_page_log";
        //一个消费者宕机后 之前分配给他的分区会重新分配给其他的消费者 实现消费者的故障容错
        //消费者组 组内的成员就可以粪蛋多个分区的压力 提高消费性能
        String groupid = "dws_traffic_source_keyword_page_view_window";

        tableEnv.executeSql("" +
                "create table page_log(" +
                "`page` map<string,string>," +
                "`ts` bigint," +
                "`rt` as TO_TIMESTAMP(from UNIXTIME(ts/1000))," +
                "WATERMARK FOR rt AS rt-INTERVAL'2'SECOND" +
                ")"+ MyKafkaUtil.getKafkaDDL(topic,groupid));

        //todo 3.过滤出搜索数据
        Table filterTable = tableEnv.sqlQuery("" +
                "select" +
                "page['item'] item," +
                "rt" +
                "from page_log" +
                "where page['last_page_id']='search'" +
                "and page['item_type']='keyword'" +
                "and page['item'] is not null");

        tableEnv.createTemporaryView("filter_table",filterTable);
        //自定义函数 把关键词拆分到对象里面 辉煌八维大数据
        tableEnv.createTemporarySystemFunction("SplitFunction", SplitFunction.class);

        //todo 4.注册UTDF & 切词 LATERAL TABLE 把对象炸裂开来
        Table splitTable = tableEnv.sqlQuery("" +
                "select" +
                "word," +
                "rt" +
                "from filter_table," +
                "LATERAL table(SplitFunction(item))");
        tableEnv.createTemporaryView("split_table",splitTable);

    }
}
